#include "Avalon/IO/PipedReader.hpp"
#include "Avalon/IO/IOException.hpp"
#include "Avalon/IO/PipedWriter.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace std;

PipedReader::PipedReader()
    : m_mutex(new recursive_mutex()),
      m_sizeRemaining(0),
      m_nextReadSize(0),
      m_destination(NULL),
      m_destinationBuffer(NULL),
      m_writer(NULL) {}

PipedReader::~PipedReader() {
  Break();
}

void PipedReader::Break() {
  lock_guard<recursive_mutex> lock(*m_mutex);
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak();
}

bool PipedReader::IsAtEndOfDataSource() {
  return false;
}

Async<int>::Ptr PipedReader::Read(void* destination, int size) {
  lock_guard<recursive_mutex> lock(*m_mutex);
  m_destination = destination;
  m_nextReadSize = size;
  if(m_sizeRemaining != 0) {
    return NewAsync(CopyOver());
  }
  if(m_writer == NULL) {
    AsyncOwner<int>::Ptr result = AsyncOwner<int>::Create();
    result->SetException(IOException("Pipe broken."));
    return result;
  }
  m_nextReadResult = AsyncOwner<int>::Create();
  return m_nextReadResult;
}

Async<int>::Ptr PipedReader::Read(Buffer* destination, int size) {
  lock_guard<recursive_mutex> lock(*m_mutex);
  m_destinationBuffer = destination;
  m_nextReadSize = size;
  if(m_sizeRemaining != 0) {
    return NewAsync(CopyOver());
  }
  if(m_writer == NULL) {
    AsyncOwner<int>::Ptr result = AsyncOwner<int>::Create();
    result->SetException(IOException("Pipe broken."));
    return result;
  }
  m_nextReadResult = AsyncOwner<int>::Create();
  return m_nextReadResult;
}

void PipedReader::Write(const void* data, int size) {
  m_data.Append(data, size);
  m_sizeRemaining += size;
  if(m_destination != NULL || m_destinationBuffer != NULL) {
    int sizeRead = CopyOver();
    AsyncOwner<int>::Ptr nextReadResult = m_nextReadResult;
    m_nextReadResult.reset();
    m_nextReadSize = 0;
    nextReadResult->SetResult(sizeRead);
  }
}

int PipedReader::CopyOver() {
  int sizeRead = min(m_nextReadSize, m_sizeRemaining);
  if(m_destinationBuffer != NULL) {
    m_destinationBuffer->Reserve(sizeRead);
    m_destination = m_destinationBuffer->GetMutableData();
  }
  const char* readPosition = m_data.GetData() +
    (m_data.GetSize() - m_sizeRemaining);
  memcpy(m_destination, readPosition, sizeRead);
  m_sizeRemaining -= sizeRead;
  m_destination = NULL;
  m_destinationBuffer = NULL;
  return sizeRead;
}

void PipedReader::Connect(PipedWriter* writer) {
  m_writer = writer;
}

void PipedReader::InternalBreak() {
  m_writer = NULL;
  if(m_nextReadResult != NULL) {
    AsyncOwner<int>::Ptr nextReadResult = m_nextReadResult;
    m_nextReadResult.reset();
    nextReadResult->SetException(IOException("Pipe broken."));
  }
}
